{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A SageMaker Workflow\n",
    "\n",
    "The pipeline that we create follows a typical Machine Learning Application pattern of pre-processing, training, evaluation, and model registration:\n",
    "\n",
    "![A typical ML Application pipeline](img/pipeline-full.png)\n",
    "\n",
    "### Create SageMaker Clients and Session\n",
    "\n",
    "First, we create a new SageMaker Session in the current AWS region. We also acquire the role arn for the session.\n",
    "\n",
    "This role arn should be the execution role arn that you set up in the Prerequisites section of this notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from botocore.exceptions import ClientError\n",
    "\n",
    "import os\n",
    "import sagemaker\n",
    "import logging\n",
    "import boto3\n",
    "import sagemaker\n",
    "import pandas as pd\n",
    "\n",
    "sess = sagemaker.Session()\n",
    "bucket = sess.default_bucket()\n",
    "role = sagemaker.get_execution_role()\n",
    "region = boto3.Session().region_name\n",
    "\n",
    "sm = boto3.Session().client(service_name=\"sagemaker\", region_name=region)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Track the Pipeline as an `Experiment`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "timestamp = int(time.time())\n",
    "\n",
    "pipeline_name = \"BERT-pipeline-{}\".format(timestamp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store pipeline_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from smexperiments.experiment import Experiment\n",
    "\n",
    "pipeline_experiment = Experiment.create(\n",
    "    experiment_name=pipeline_name,\n",
    "    description=\"Amazon Customer Reviews BERT Pipeline Experiment\",\n",
    "    sagemaker_boto_client=sm,\n",
    ")\n",
    "\n",
    "pipeline_experiment_name = pipeline_experiment.experiment_name\n",
    "print(\"Pipeline experiment name: {}\".format(pipeline_experiment_name))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store pipeline_experiment_name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create the `Trial`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from smexperiments.trial import Trial\n",
    "\n",
    "pipeline_trial = Trial.create(\n",
    "    trial_name=\"trial-{}\".format(timestamp), experiment_name=pipeline_experiment_name, sagemaker_boto_client=sm\n",
    ")\n",
    "\n",
    "pipeline_trial_name = pipeline_trial.trial_name\n",
    "print(\"Trial name: {}\".format(pipeline_trial_name))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store pipeline_trial_name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Define Parameters to Parametrize Pipeline Execution\n",
    "\n",
    "We define Workflow Parameters by which we can parametrize our Pipeline and vary the values injected and used in Pipeline executions and schedules without having to modify the Pipeline definition.\n",
    "\n",
    "The supported parameter types include:\n",
    "\n",
    "* `ParameterString` - representing a `str` Python type\n",
    "* `ParameterInteger` - representing an `int` Python type\n",
    "* `ParameterFloat` - representing a `float` Python type\n",
    "\n",
    "These parameters support providing a default value, which can be overridden on pipeline execution. The default value specified should be an instance of the type of the parameter.\n",
    "\n",
    "The parameters defined in this workflow below include:\n",
    "\n",
    "* `processing_instance_type` - The `ml.*` instance type of the processing job.\n",
    "* `processing_instance_count` - The instance count of the processing job. For illustrative purposes only: 1 is the only value that makes sense here.\n",
    "* `train_instance_type` - The `ml.*` instance type of the training job.\n",
    "* `model_approval_status` - What approval status to register the trained model with for CI/CD purposes. Defaults to \"PendingManualApproval\". (NOTE: not available in service yet)\n",
    "* `input_data` - The URL location of the input data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Pipeline Parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.parameters import (\n",
    "    ParameterInteger,\n",
    "    ParameterString,\n",
    "    ParameterFloat,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Experiment Parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r pipeline_experiment_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "exp_name = ParameterString(\n",
    "    name=\"ExperimentName\",\n",
    "    default_value=pipeline_experiment_name,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Processing Step Parameters"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Define a Processing Step for Feature Engineering](img/pipeline-2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](img/prepare_dataset_bert.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_input_data_s3_uri = \"s3://{}/amazon-reviews-pds/tsv/\".format(bucket)\n",
    "print(raw_input_data_s3_uri)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!aws s3 ls $raw_input_data_s3_uri"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "timestamp = int(time.time())\n",
    "\n",
    "input_data = ParameterString(\n",
    "    name=\"InputData\",\n",
    "    default_value=raw_input_data_s3_uri,\n",
    ")\n",
    "\n",
    "processing_instance_count = ParameterInteger(name=\"ProcessingInstanceCount\", default_value=1)\n",
    "\n",
    "processing_instance_type = ParameterString(name=\"ProcessingInstanceType\", default_value=\"ml.c5.2xlarge\")\n",
    "\n",
    "max_seq_length = ParameterInteger(\n",
    "    name=\"MaxSeqLength\",\n",
    "    default_value=64,\n",
    ")\n",
    "\n",
    "balance_dataset = ParameterString(\n",
    "    name=\"BalanceDataset\",\n",
    "    default_value=\"True\",\n",
    ")\n",
    "\n",
    "train_split_percentage = ParameterFloat(\n",
    "    name=\"TrainSplitPercentage\",\n",
    "    default_value=0.90,\n",
    ")\n",
    "\n",
    "validation_split_percentage = ParameterFloat(\n",
    "    name=\"ValidationSplitPercentage\",\n",
    "    default_value=0.05,\n",
    ")\n",
    "\n",
    "test_split_percentage = ParameterFloat(\n",
    "    name=\"TestSplitPercentage\",\n",
    "    default_value=0.05,\n",
    ")\n",
    "\n",
    "feature_store_offline_prefix = ParameterString(\n",
    "    name=\"FeatureStoreOfflinePrefix\",\n",
    "    default_value=\"reviews-feature-store-\" + str(timestamp),\n",
    ")\n",
    "\n",
    "feature_group_name = ParameterString(name=\"FeatureGroupName\", default_value=\"reviews-feature-group-\" + str(timestamp))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!pygmentize ./preprocess-scikit-text-to-bert-feature-store.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create an instance of an `SKLearnProcessor` processor and we use that in our `ProcessingStep`.\n",
    "\n",
    "We also specify the `framework_version` we will use throughout.\n",
    "\n",
    "Note the `processing_instance_type` and `processing_instance_count` parameters that used by the processor instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "\n",
    "processor = SKLearnProcessor(\n",
    "    framework_version=\"0.23-1\",\n",
    "    role=role,\n",
    "    instance_type=processing_instance_type,\n",
    "    instance_count=processing_instance_count,\n",
    "    env={\"AWS_DEFAULT_REGION\": region},\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
    "from sagemaker.workflow.steps import ProcessingStep\n",
    "\n",
    "processing_inputs = [\n",
    "    ProcessingInput(\n",
    "        input_name=\"raw-input-data\",\n",
    "        source=input_data,\n",
    "        destination=\"/opt/ml/processing/input/data/\",\n",
    "        s3_data_distribution_type=\"ShardedByS3Key\",\n",
    "    )\n",
    "]\n",
    "\n",
    "processing_outputs = [\n",
    "    ProcessingOutput(\n",
    "        output_name=\"bert-train\",\n",
    "        s3_upload_mode=\"EndOfJob\",\n",
    "        source=\"/opt/ml/processing/output/bert/train\",\n",
    "    ),\n",
    "    ProcessingOutput(\n",
    "        output_name=\"bert-validation\",\n",
    "        s3_upload_mode=\"EndOfJob\",\n",
    "        source=\"/opt/ml/processing/output/bert/validation\",\n",
    "    ),\n",
    "    ProcessingOutput(\n",
    "        output_name=\"bert-test\",\n",
    "        s3_upload_mode=\"EndOfJob\",\n",
    "        source=\"/opt/ml/processing/output/bert/test\",\n",
    "    ),\n",
    "]\n",
    "\n",
    "processing_step = ProcessingStep(\n",
    "    name=\"Processing\",\n",
    "    code=\"preprocess-scikit-text-to-bert-feature-store.py\",\n",
    "    processor=processor,\n",
    "    inputs=processing_inputs,\n",
    "    outputs=processing_outputs,\n",
    "    job_arguments=[\n",
    "        \"--train-split-percentage\",\n",
    "        str(train_split_percentage.default_value),\n",
    "        \"--validation-split-percentage\",\n",
    "        str(validation_split_percentage.default_value),\n",
    "        \"--test-split-percentage\",\n",
    "        str(test_split_percentage.default_value),\n",
    "        \"--max-seq-length\",\n",
    "        str(max_seq_length.default_value),\n",
    "        \"--balance-dataset\",\n",
    "        str(balance_dataset.default_value),\n",
    "        \"--feature-store-offline-prefix\",\n",
    "        str(feature_store_offline_prefix.default_value),\n",
    "        \"--feature-group-name\",\n",
    "        str(feature_group_name.default_value),\n",
    "    ],\n",
    ")\n",
    "\n",
    "print(processing_step)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, we use the processor instance to construct a `ProcessingStep`, along with the input and output channels and the code that will be executed when the pipeline invokes pipeline execution. This is very similar to a processor instance's `run` method, for those familiar with the existing Python SDK.\n",
    "\n",
    "Note the `input_data` parameters passed into `ProcessingStep` as the input data of the step itself. This input data will be used by the processor instance when it is run.\n",
    "\n",
    "Also, take note the `\"bert-train\"`, `\"bert-validation\"` and `\"bert-test\"` named channels specified in the output configuration for the processing job. Such step `Properties` can be used in subsequent steps and will resolve to their runtime values at execution. In particular, we'll call out this usage when we define our training step."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Train Step"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Define a Training Step to Train a Model](img/pipeline-3.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_instance_type = ParameterString(name=\"TrainInstanceType\", default_value=\"ml.c5.9xlarge\")\n",
    "\n",
    "train_instance_count = ParameterInteger(name=\"TrainInstanceCount\", default_value=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Training Hyper-Parameters\n",
    "Note that `max_seq_length` is re-used from the processing hyper-parameters above"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "epochs = ParameterInteger(name=\"Epochs\", default_value=1)\n",
    "\n",
    "learning_rate = ParameterFloat(name=\"LearningRate\", default_value=0.00001)\n",
    "\n",
    "epsilon = ParameterFloat(name=\"Epsilon\", default_value=0.00000001)\n",
    "\n",
    "train_batch_size = ParameterInteger(name=\"TrainBatchSize\", default_value=128)\n",
    "\n",
    "validation_batch_size = ParameterInteger(name=\"ValidationBatchSize\", default_value=128)\n",
    "\n",
    "test_batch_size = ParameterInteger(name=\"TestBatchSize\", default_value=128)\n",
    "\n",
    "train_steps_per_epoch = ParameterInteger(name=\"TrainStepsPerEpoch\", default_value=50)\n",
    "\n",
    "validation_steps = ParameterInteger(name=\"ValidationSteps\", default_value=50)\n",
    "\n",
    "test_steps = ParameterInteger(name=\"TestSteps\", default_value=50)\n",
    "\n",
    "train_volume_size = ParameterInteger(name=\"TrainVolumeSize\", default_value=1024)\n",
    "\n",
    "use_xla = ParameterString(\n",
    "    name=\"UseXLA\",\n",
    "    default_value=\"True\",\n",
    ")\n",
    "\n",
    "use_amp = ParameterString(\n",
    "    name=\"UseAMP\",\n",
    "    default_value=\"True\",\n",
    ")\n",
    "\n",
    "freeze_bert_layer = ParameterString(\n",
    "    name=\"FreezeBERTLayer\",\n",
    "    default_value=\"False\",\n",
    ")\n",
    "\n",
    "enable_sagemaker_debugger = ParameterString(\n",
    "    name=\"EnableSageMakerDebugger\",\n",
    "    default_value=\"False\",\n",
    ")\n",
    "\n",
    "enable_checkpointing = ParameterString(\n",
    "    name=\"EnableCheckpointing\",\n",
    "    default_value=\"False\",\n",
    ")\n",
    "\n",
    "enable_tensorboard = ParameterString(\n",
    "    name=\"EnableTensorboard\",\n",
    "    default_value=\"False\",\n",
    ")\n",
    "\n",
    "input_mode = ParameterString(\n",
    "    name=\"InputMode\",\n",
    "    default_value=\"File\",\n",
    ")\n",
    "\n",
    "run_validation = ParameterString(\n",
    "    name=\"RunValidation\",\n",
    "    default_value=\"True\",\n",
    ")\n",
    "\n",
    "run_test = ParameterString(\n",
    "    name=\"RunTest\",\n",
    "    default_value=\"False\",\n",
    ")\n",
    "\n",
    "run_sample_predictions = ParameterString(\n",
    "    name=\"RunSamplePredictions\",\n",
    "    default_value=\"False\",\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Metrics To Track Model Performance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "metrics_definitions = [\n",
    "    {\"Name\": \"train:loss\", \"Regex\": \"loss: ([0-9\\\\.]+)\"},\n",
    "    {\"Name\": \"train:accuracy\", \"Regex\": \"accuracy: ([0-9\\\\.]+)\"},\n",
    "    {\"Name\": \"validation:loss\", \"Regex\": \"val_loss: ([0-9\\\\.]+)\"},\n",
    "    {\"Name\": \"validation:accuracy\", \"Regex\": \"val_accuracy: ([0-9\\\\.]+)\"},\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!pygmentize src/tf_bert_reviews.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Setup Debugger and Profiler\n",
    "Define Debugger Rules as described here:  https://docs.aws.amazon.com/sagemaker/latest/dg/debugger-built-in-rules.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.debugger import Rule, ProfilerRule, rule_configs\n",
    "from sagemaker.debugger import DebuggerHookConfig\n",
    "from sagemaker.debugger import ProfilerConfig, FrameworkProfile\n",
    "\n",
    "debugger_hook_config = DebuggerHookConfig(\n",
    "    s3_output_path=\"s3://{}\".format(bucket),\n",
    ")\n",
    "\n",
    "profiler_config = ProfilerConfig(\n",
    "    system_monitor_interval_millis=500,\n",
    "    framework_profile_params=FrameworkProfile(local_path=\"/opt/ml/output/profiler/\", start_step=5, num_steps=10),\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rules = [ProfilerRule.sagemaker(rule_configs.ProfilerReport())]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Define a Training Step to Train a Model\n",
    "\n",
    "We configure an Estimator and the input dataset. A typical training script loads data from the input channels, configures training with hyperparameters, trains a model, and saves a model to `model_dir` so that it can be hosted later.\n",
    "\n",
    "We also specify the model path where the models from training will be saved.\n",
    "\n",
    "Note the `train_instance_type` parameter passed may be also used and passed into other places in the pipeline. In this case, the `train_instance_type` is passed into the estimator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.tensorflow import TensorFlow\n",
    "\n",
    "estimator = TensorFlow(\n",
    "    entry_point=\"tf_bert_reviews.py\",\n",
    "    source_dir=\"src\",\n",
    "    role=role,\n",
    "    instance_count=train_instance_count,  # Make sure you have at least this number of input files or the ShardedByS3Key distibution strategy will fail the job due to no data available\n",
    "    instance_type=train_instance_type,\n",
    "    volume_size=train_volume_size,\n",
    "    py_version=\"py37\",\n",
    "    framework_version=\"2.3.1\",\n",
    "    hyperparameters={\n",
    "        \"epochs\": epochs,\n",
    "        \"learning_rate\": learning_rate,\n",
    "        \"epsilon\": epsilon,\n",
    "        \"train_batch_size\": train_batch_size,\n",
    "        \"validation_batch_size\": validation_batch_size,\n",
    "        \"test_batch_size\": test_batch_size,\n",
    "        \"train_steps_per_epoch\": train_steps_per_epoch,\n",
    "        \"validation_steps\": validation_steps,\n",
    "        \"test_steps\": test_steps,\n",
    "        \"use_xla\": use_xla,\n",
    "        \"use_amp\": use_amp,\n",
    "        \"max_seq_length\": max_seq_length,\n",
    "        \"freeze_bert_layer\": freeze_bert_layer,\n",
    "        \"enable_sagemaker_debugger\": enable_sagemaker_debugger,\n",
    "        \"enable_checkpointing\": enable_checkpointing,\n",
    "        \"enable_tensorboard\": enable_tensorboard,\n",
    "        \"run_validation\": run_validation,\n",
    "        \"run_test\": run_test,\n",
    "        \"run_sample_predictions\": run_sample_predictions,\n",
    "    },\n",
    "    input_mode=input_mode,\n",
    "    metric_definitions=metrics_definitions,\n",
    "    debugger_hook_config=debugger_hook_config,\n",
    "    profiler_config=profiler_config,\n",
    "    rules=rules,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Setup Pipeline Step Caching\n",
    "Cache pipeline steps for a duration of time using [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601#Durations) format.  \n",
    "\n",
    "More details on SageMaker Pipeline step caching here:  https://docs.aws.amazon.com/sagemaker/latest/dg/pipelines-caching.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.steps import CacheConfig\n",
    "\n",
    "cache_config = CacheConfig(enable_caching=True, expire_after=\"PT1H\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Configure Training Step"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, we use the estimator instance to construct a `TrainingStep` as well as the `Properties` of the prior `ProcessingStep` used as input in the `TrainingStep` inputs and the code that will be executed when the pipeline invokes pipeline execution. This is very similar to an estimator's `fit` method, for those familiar with the existing Python SDK.\n",
    "\n",
    "In particular, we pass in the `S3Uri` of the `\"train\"`, `\"validation\"` and `\"test\"` output channel to the `TrainingStep`. The `properties` attribute of a Workflow step match the object model of the corresponding response of a describe call. These properties can be referenced as placeholder values and are resolved, or filled in, at runtime. For example, the `ProcessingStep` `properties` attribute matches the object model of the [DescribeProcessingJob](https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DescribeProcessingJob.html) response object."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.inputs import TrainingInput\n",
    "from sagemaker.workflow.steps import TrainingStep\n",
    "\n",
    "training_step = TrainingStep(\n",
    "    name=\"Train\",\n",
    "    estimator=estimator,\n",
    "    inputs={\n",
    "        \"train\": TrainingInput(\n",
    "            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs[\"bert-train\"].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "        \"validation\": TrainingInput(\n",
    "            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs[\"bert-validation\"].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "        \"test\": TrainingInput(\n",
    "            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs[\"bert-test\"].S3Output.S3Uri,\n",
    "            content_type=\"text/csv\",\n",
    "        ),\n",
    "    },\n",
    "    cache_config=cache_config,\n",
    ")\n",
    "\n",
    "print(training_step)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Evaluation Step"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Define a Model Evaluation Step to Evaluate the Trained Model](img/pipeline-4.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, we develop an evaluation script that will be specified in a Processing step that will perform the model evaluation.\n",
    "\n",
    "The evaluation script `evaluation.py` takes the trained model and the test dataset as input, and produces a JSON file containing classification evaluation metrics such as accuracy.\n",
    "\n",
    "After pipeline execution, we will examine the resulting `evaluation.json` for analysis.\n",
    "\n",
    "The evaluation script:\n",
    "\n",
    "* loads in the model\n",
    "* reads in the test data\n",
    "* issues a bunch of predictions against the test data\n",
    "* builds a classification report, including accuracy\n",
    "* saves the evaluation report to the evaluation directory\n",
    "\n",
    "Next, we create an instance of a `ScriptProcessor` processor and we use that in our `ProcessingStep`.\n",
    "\n",
    "Note the `processing_instance_type` parameter passed into the processor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "\n",
    "evaluation_processor = SKLearnProcessor(\n",
    "    framework_version=\"0.23-1\",\n",
    "    role=role,\n",
    "    instance_type=processing_instance_type,\n",
    "    instance_count=processing_instance_count,\n",
    "    env={\"AWS_DEFAULT_REGION\": region},\n",
    "    max_runtime_in_seconds=7200,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!pygmentize evaluate_model_metrics.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We use the processor instance to construct a `ProcessingStep`, along with the input and output channels and the code that will be executed when the pipeline invokes pipeline execution. This is very similar to a processor instance's `run` method, for those familiar with the existing Python SDK.\n",
    "\n",
    "The `TrainingStep` and `ProcessingStep` `properties` attribute matches the object model of the [DescribeTrainingJob](https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DescribeTrainingJob.html) and  [DescribeProcessingJob](https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DescribeProcessingJob.html) response objects, respectively."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.properties import PropertyFile\n",
    "\n",
    "evaluation_report = PropertyFile(name=\"EvaluationReport\", output_name=\"metrics\", path=\"evaluation.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluation_step = ProcessingStep(\n",
    "    name=\"EvaluateModel\",\n",
    "    processor=evaluation_processor,\n",
    "    code=\"evaluate_model_metrics.py\",\n",
    "    inputs=[\n",
    "        ProcessingInput(\n",
    "            source=training_step.properties.ModelArtifacts.S3ModelArtifacts,\n",
    "            destination=\"/opt/ml/processing/input/model\",\n",
    "        ),\n",
    "        ProcessingInput(\n",
    "            source=processing_step.properties.ProcessingInputs[\"raw-input-data\"].S3Input.S3Uri,\n",
    "            destination=\"/opt/ml/processing/input/data\",\n",
    "        ),\n",
    "    ],\n",
    "    outputs=[\n",
    "        ProcessingOutput(\n",
    "            output_name=\"metrics\", s3_upload_mode=\"EndOfJob\", source=\"/opt/ml/processing/output/metrics/\"\n",
    "        ),\n",
    "    ],\n",
    "    job_arguments=[\n",
    "        \"--max-seq-length\",\n",
    "        str(max_seq_length.default_value),\n",
    "    ],\n",
    "    property_files=[evaluation_report],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.model_metrics import MetricsSource, ModelMetrics\n",
    "\n",
    "model_metrics = ModelMetrics(\n",
    "    model_statistics=MetricsSource(\n",
    "        s3_uri=\"{}/evaluation.json\".format(\n",
    "            evaluation_step.arguments[\"ProcessingOutputConfig\"][\"Outputs\"][0][\"S3Output\"][\"S3Uri\"]\n",
    "        ),\n",
    "        content_type=\"application/json\",\n",
    "    )\n",
    ")\n",
    "\n",
    "print(model_metrics)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Register Model Step\n",
    "\n",
    "![](img/pipeline-5.png)\n",
    "\n",
    "We use the estimator instance that was used for the training step to construct an instance of `RegisterModel`. The result of executing `RegisterModel` in a pipeline is a Model Package. A Model Package is a reusable model artifacts abstraction that packages all ingredients necessary for inference. Primarily, it consists of an inference specification that defines the inference image to use along with an optional model weights location.\n",
    "\n",
    "A Model Package Group is a collection of Model Packages. You can create a Model Package Group for a specific ML business problem, and you can keep adding versions/model packages into it. Typically, we expect customers to create a ModelPackageGroup for a SageMaker Workflow Pipeline so that they can keep adding versions/model packages to the group for every Workflow Pipeline run.\n",
    "\n",
    "The construction of `RegisterModel` is very similar to an estimator instance's `register` method, for those familiar with the existing Python SDK.\n",
    "\n",
    "In particular, we pass in the `S3ModelArtifacts` from the `TrainingStep`, `step_train` properties. The `TrainingStep` `properties` attribute matches the object model of the [DescribeTrainingJob](https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DescribeTrainingJob.html) response object.\n",
    "\n",
    "Of note, we provided a specific model package group name which we will use in the Model Registry and CI/CD work later on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_approval_status = ParameterString(name=\"ModelApprovalStatus\", default_value=\"PendingManualApproval\")\n",
    "\n",
    "deploy_instance_type = ParameterString(name=\"DeployInstanceType\", default_value=\"ml.m4.xlarge\")\n",
    "\n",
    "deploy_instance_count = ParameterInteger(name=\"DeployInstanceCount\", default_value=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_package_group_name = f\"BERT-Reviews-{timestamp}\"\n",
    "\n",
    "print(model_package_group_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inference_image_uri = sagemaker.image_uris.retrieve(\n",
    "    framework=\"tensorflow\",\n",
    "    region=region,\n",
    "    version=\"2.3.1\",\n",
    "    py_version=\"py37\",\n",
    "    instance_type=deploy_instance_type,\n",
    "    image_scope=\"inference\",\n",
    ")\n",
    "print(inference_image_uri)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.step_collections import RegisterModel\n",
    "\n",
    "register_step = RegisterModel(\n",
    "    name=\"RegisterModel\",\n",
    "    #    entry_point='inference.py', # Adds a Repack Step:  https://github.com/aws/sagemaker-python-sdk/blob/01c6ee3a9ec1831e935e86df58cf70bc92ed1bbe/src/sagemaker/workflow/_utils.py#L44\n",
    "    #    source_dir='src',\n",
    "    estimator=estimator,\n",
    "    image_uri=inference_image_uri,  # we have to specify, by default it's using training image\n",
    "    model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,\n",
    "    content_types=[\"application/jsonlines\"],\n",
    "    response_types=[\"application/jsonlines\"],\n",
    "    inference_instances=[deploy_instance_type],\n",
    "    transform_instances=[\"ml.m4.xlarge\"],\n",
    "    model_package_group_name=model_package_group_name,\n",
    "    approval_status=model_approval_status,\n",
    "    model_metrics=model_metrics,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create Model for Deployment Step\n",
    "\n",
    "![](img/pipeline-5.png)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.model import Model\n",
    "\n",
    "model_name = \"bert-model-{}\".format(timestamp)\n",
    "\n",
    "model = Model(\n",
    "    name=model_name,\n",
    "    image_uri=inference_image_uri,\n",
    "    model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,\n",
    "    sagemaker_session=sess,\n",
    "    role=role,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.inputs import CreateModelInput\n",
    "\n",
    "create_inputs = CreateModelInput(\n",
    "    instance_type=deploy_instance_type,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.steps import CreateModelStep\n",
    "\n",
    "create_step = CreateModelStep(\n",
    "    name=\"CreateModel\",\n",
    "    model=model,\n",
    "    inputs=create_inputs,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Define a Condition Step to Check Accuracy and Conditionally Register Model\n",
    "\n",
    "![](img/pipeline-6.png)\n",
    "\n",
    "Finally, we'd like to only register this model if the accuracy of the model, as determined by our evaluation step `step_eval`, exceeded some value. A `ConditionStep` allows for pipelines to support conditional execution in the pipeline DAG based on conditions of step properties. \n",
    "\n",
    "Below, we:\n",
    "\n",
    "* define a `ConditionGreaterThan` on the accuracy value found in the output of the evaluation step, `step_eval`.\n",
    "* use the condition in the list of conditions in a `ConditionStep`\n",
    "* pass the `RegisterModel` step collection into the `if_steps` of the `ConditionStep`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "min_accuracy_value = ParameterFloat(name=\"MinAccuracyValue\", default_value=0.01)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo\n",
    "from sagemaker.workflow.condition_step import (\n",
    "    ConditionStep,\n",
    "    JsonGet,\n",
    ")\n",
    "\n",
    "minimum_accuracy_condition = ConditionGreaterThanOrEqualTo(\n",
    "    left=JsonGet(\n",
    "        step=evaluation_step,\n",
    "        property_file=evaluation_report,\n",
    "        json_path=\"metrics.accuracy.value\",\n",
    "    ),\n",
    "    right=min_accuracy_value,  # accuracy\n",
    ")\n",
    "\n",
    "minimum_accuracy_condition_step = ConditionStep(\n",
    "    name=\"AccuracyCondition\",\n",
    "    conditions=[minimum_accuracy_condition],\n",
    "    if_steps=[register_step, create_step],  # success, continue with model registration\n",
    "    else_steps=[],  # fail, end the pipeline\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Define a Pipeline of Parameters, Steps, and Conditions\n",
    "\n",
    "Let's tie it all up into a workflow pipeline so we can execute it, and even schedule it.\n",
    "\n",
    "A pipeline requires a `name`, `parameters`, and `steps`. Names must be unique within an `(account, region)` pair so we tack on the timestamp to the name.\n",
    "\n",
    "Note:\n",
    "\n",
    "* All the parameters used in the definitions must be present.\n",
    "* Steps passed into the pipeline need not be in the order of execution. The SageMaker Workflow service will resolve the _data dependency_ DAG as steps the execution complete.\n",
    "* Steps must be unique to either pipeline step list or a single condition step if/else list."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r pipeline_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.pipeline import Pipeline\n",
    "\n",
    "pipeline = Pipeline(\n",
    "    name=pipeline_name,\n",
    "    parameters=[\n",
    "        input_data,\n",
    "        processing_instance_count,\n",
    "        processing_instance_type,\n",
    "        max_seq_length,\n",
    "        balance_dataset,\n",
    "        train_split_percentage,\n",
    "        validation_split_percentage,\n",
    "        test_split_percentage,\n",
    "        feature_store_offline_prefix,\n",
    "        feature_group_name,\n",
    "        train_instance_type,\n",
    "        train_instance_count,\n",
    "        epochs,\n",
    "        learning_rate,\n",
    "        epsilon,\n",
    "        train_batch_size,\n",
    "        validation_batch_size,\n",
    "        test_batch_size,\n",
    "        train_steps_per_epoch,\n",
    "        validation_steps,\n",
    "        test_steps,\n",
    "        train_volume_size,\n",
    "        use_xla,\n",
    "        use_amp,\n",
    "        freeze_bert_layer,\n",
    "        enable_sagemaker_debugger,\n",
    "        enable_checkpointing,\n",
    "        enable_tensorboard,\n",
    "        input_mode,\n",
    "        run_validation,\n",
    "        run_test,\n",
    "        run_sample_predictions,\n",
    "        min_accuracy_value,\n",
    "        model_approval_status,\n",
    "        deploy_instance_type,\n",
    "        deploy_instance_count,\n",
    "    ],\n",
    "    steps=[processing_step, training_step, evaluation_step, minimum_accuracy_condition_step],\n",
    "    sagemaker_session=sess,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's examine the Json of the pipeline definition that meets the SageMaker Workflow Pipeline DSL specification.\n",
    "\n",
    "By examining the definition, we're also confirming that the pipeline was well-defined, and that the parameters and step properties resolve correctly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import json\n",
    "from pprint import pprint\n",
    "\n",
    "definition = json.loads(pipeline.definition())\n",
    "\n",
    "pprint(definition)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit the pipeline to SageMaker and start execution\n",
    "\n",
    "Let's submit our pipeline definition to the workflow service. The role passed in will be used by the workflow service to create all the jobs defined in the steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(pipeline_experiment_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ignore the `WARNING` below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = pipeline.create(role_arn=role)\n",
    "\n",
    "pipeline_arn = response[\"PipelineArn\"]\n",
    "print(pipeline_arn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll start the pipeline, accepting all the default parameters.\n",
    "\n",
    "Values can also be passed into these pipeline parameters on starting of the pipeline, and will be covered later. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution = pipeline.start(\n",
    "    parameters=dict(\n",
    "        InputData=raw_input_data_s3_uri,\n",
    "        ProcessingInstanceCount=1,\n",
    "        ProcessingInstanceType=\"ml.c5.2xlarge\",\n",
    "        MaxSeqLength=64,\n",
    "        BalanceDataset=\"True\",\n",
    "        TrainSplitPercentage=0.9,\n",
    "        ValidationSplitPercentage=0.05,\n",
    "        TestSplitPercentage=0.05,\n",
    "        FeatureStoreOfflinePrefix=\"reviews-feature-store-\" + str(timestamp),\n",
    "        FeatureGroupName=\"reviews-feature-group-\" + str(timestamp),\n",
    "        LearningRate=0.000012,\n",
    "        TrainInstanceType=\"ml.c5.9xlarge\",\n",
    "        TrainInstanceCount=1,\n",
    "        Epochs=1,\n",
    "        Epsilon=0.00000001,\n",
    "        TrainBatchSize=128,\n",
    "        ValidationBatchSize=128,\n",
    "        TestBatchSize=128,\n",
    "        TrainStepsPerEpoch=50,\n",
    "        ValidationSteps=50,\n",
    "        TestSteps=50,\n",
    "        TrainVolumeSize=1024,\n",
    "        UseXLA=\"True\",\n",
    "        UseAMP=\"True\",\n",
    "        FreezeBERTLayer=\"False\",\n",
    "        EnableSageMakerDebugger=\"False\",\n",
    "        EnableCheckpointing=\"False\",\n",
    "        EnableTensorboard=\"False\",\n",
    "        InputMode=\"File\",\n",
    "        RunValidation=\"True\",\n",
    "        RunTest=\"False\",\n",
    "        RunSamplePredictions=\"False\",\n",
    "        MinAccuracyValue=0.01,\n",
    "        ModelApprovalStatus=\"PendingManualApproval\",\n",
    "        DeployInstanceType=\"ml.m4.xlarge\",\n",
    "        DeployInstanceCount=1,\n",
    "    )\n",
    ")\n",
    "\n",
    "print(execution.arn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Workflow Operations: examining and waiting for pipeline execution\n",
    "\n",
    "Now we describe execution instance and list the steps in the execution to find out more about the execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pprint import pprint\n",
    "\n",
    "execution_run = execution.describe()\n",
    "pprint(execution_run)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Add Execution Run as Trial to Experiments"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution_run_name = execution_run[\"PipelineExecutionDisplayName\"]\n",
    "print(execution_run_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_execution_arn = execution_run[\"PipelineExecutionArn\"]\n",
    "print(pipeline_execution_arn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# List Execution Steps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "# Giving the first step time to start up\n",
    "time.sleep(30)\n",
    "\n",
    "execution.list_steps()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Wait for the Pipeline to Complete\n",
    "\n",
    "# _Note: If this cell errors out with `WaiterError: Waiter PipelineExecutionComplete failed: Max attempts exceeded`, just re-run it and keep waiting._"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r pipeline_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "import time\n",
    "from pprint import pprint\n",
    "\n",
    "executions_response = sm.list_pipeline_executions(PipelineName=pipeline_name)[\"PipelineExecutionSummaries\"]\n",
    "pipeline_execution_status = executions_response[0][\"PipelineExecutionStatus\"]\n",
    "print(pipeline_execution_status)\n",
    "\n",
    "while pipeline_execution_status == \"Executing\":\n",
    "    try:\n",
    "        executions_response = sm.list_pipeline_executions(PipelineName=pipeline_name)[\"PipelineExecutionSummaries\"]\n",
    "        pipeline_execution_status = executions_response[0][\"PipelineExecutionStatus\"]\n",
    "    #        print('Executions for our pipeline...')\n",
    "    #        print(pipeline_execution_status)\n",
    "    except Exception as e:\n",
    "        print(\"Please wait...\")\n",
    "        time.sleep(30)\n",
    "\n",
    "pprint(executions_response)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Wait for the Pipeline ^^ Above ^^ to Complete\n",
    "\n",
    "# _Note: If this cell errors out with `WaiterError: Waiter PipelineExecutionComplete failed: Max attempts exceeded`, just re-run it and keep waiting._"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_execution_status = executions_response[0][\"PipelineExecutionStatus\"]\n",
    "print(pipeline_execution_status)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_execution_arn = executions_response[0][\"PipelineExecutionArn\"]\n",
    "print(pipeline_execution_arn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can list the execution steps to check out the status and artifacts:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# List Pipeline Execution Steps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_execution_status = executions_response[0][\"PipelineExecutionStatus\"]\n",
    "print(pipeline_execution_status)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pprint import pprint\n",
    "\n",
    "steps = sm.list_pipeline_execution_steps(PipelineExecutionArn=pipeline_execution_arn)\n",
    "\n",
    "pprint(steps)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# List All Artifacts Generated By The Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "processing_job_name = None\n",
    "training_job_name = None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "from sagemaker.lineage.visualizer import LineageTableVisualizer\n",
    "\n",
    "viz = LineageTableVisualizer(sagemaker.session.Session())\n",
    "\n",
    "for execution_step in reversed(steps[\"PipelineExecutionSteps\"]):\n",
    "    print(execution_step)\n",
    "    # We are doing this because there appears to be a bug of this LineageTableVisualizer handling the Processing Step\n",
    "    if execution_step[\"StepName\"] == \"Processing\":\n",
    "        processing_job_name = execution_step[\"Metadata\"][\"ProcessingJob\"][\"Arn\"].split(\"/\")[-1]\n",
    "        print(processing_job_name)\n",
    "        display(viz.show(processing_job_name=processing_job_name))\n",
    "    elif execution_step[\"StepName\"] == \"Train\":\n",
    "        training_job_name = execution_step[\"Metadata\"][\"TrainingJob\"][\"Arn\"].split(\"/\")[-1]\n",
    "        print(training_job_name)\n",
    "        display(viz.show(training_job_name=training_job_name))\n",
    "    else:\n",
    "        display(viz.show(pipeline_execution_step=execution_step))\n",
    "        time.sleep(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Track Additional Parameters in our Experiment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# -aws-processing-job is the default name assigned by ProcessingJob\n",
    "processing_job_tc = \"{}-aws-processing-job\".format(processing_job_name)\n",
    "print(processing_job_tc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r pipeline_trial_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(pipeline_trial_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = sm.associate_trial_component(TrialComponentName=processing_job_tc, TrialName=pipeline_trial_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# -aws-training-job is the default name assigned by TrainingJob\n",
    "training_job_tc = \"{}-aws-training-job\".format(training_job_name)\n",
    "print(training_job_tc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = sm.associate_trial_component(TrialComponentName=training_job_tc, TrialName=pipeline_trial_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from smexperiments import tracker\n",
    "\n",
    "processing_job_tracker = tracker.Tracker.load(trial_component_name=processing_job_tc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "processing_job_tracker.log_parameters(\n",
    "    {\n",
    "        \"balance_dataset\": str(balance_dataset),\n",
    "    }\n",
    ")\n",
    "\n",
    "# must save after logging\n",
    "processing_job_tracker.trial_component.save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "processing_job_tracker.log_parameters(\n",
    "    {\n",
    "        \"train_split_percentage\": str(train_split_percentage),\n",
    "    }\n",
    ")\n",
    "\n",
    "# must save after logging\n",
    "processing_job_tracker.trial_component.save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "processing_job_tracker.log_parameters(\n",
    "    {\n",
    "        \"validation_split_percentage\": str(validation_split_percentage),\n",
    "    }\n",
    ")\n",
    "\n",
    "# must save after logging\n",
    "processing_job_tracker.trial_component.save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "processing_job_tracker.log_parameters(\n",
    "    {\n",
    "        \"test_split_percentage\": str(test_split_percentage),\n",
    "    }\n",
    ")\n",
    "\n",
    "# must save after logging\n",
    "processing_job_tracker.trial_component.save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "processing_job_tracker.log_parameters(\n",
    "    {\n",
    "        \"max_seq_length\": str(max_seq_length),\n",
    "    }\n",
    ")\n",
    "\n",
    "# must save after logging\n",
    "processing_job_tracker.trial_component.save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.sleep(5)  # avoid throttling exception\n",
    "\n",
    "processing_job_tracker.log_parameters(\n",
    "    {\n",
    "        \"feature_store_offline_prefix\": str(feature_store_offline_prefix),\n",
    "    }\n",
    ")\n",
    "\n",
    "# must save after logging\n",
    "processing_job_tracker.trial_component.save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.sleep(5)  # avoid throttling exception\n",
    "\n",
    "processing_job_tracker.log_parameters(\n",
    "    {\n",
    "        \"feature_group_name\": str(feature_group_name),\n",
    "    }\n",
    ")\n",
    "\n",
    "# must save after logging\n",
    "processing_job_tracker.trial_component.save()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analyze Experiment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.analytics import ExperimentAnalytics\n",
    "\n",
    "time.sleep(30)  # avoid throttling exception\n",
    "\n",
    "import pandas as pd\n",
    "\n",
    "pd.set_option(\"max_colwidth\", 500)\n",
    "\n",
    "experiment_analytics = ExperimentAnalytics(\n",
    "    experiment_name=pipeline_experiment_name,\n",
    ")\n",
    "\n",
    "experiment_analytics_df = experiment_analytics.dataframe()\n",
    "experiment_analytics_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analyze SageMaker Debugger Results "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "restored_estimator = sagemaker.estimator.Estimator.attach(training_job_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.display import display, HTML\n",
    "\n",
    "display(\n",
    "    HTML(\n",
    "        '<b>Review <a target=\"blank\" href=\"https://s3.console.aws.amazon.com/s3/buckets/{}?prefix={}/\">S3 Debugger Output Data</a></b>'.format(\n",
    "            bucket, restored_estimator.base_job_name\n",
    "        )\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Download SageMaker Debugger Profiling Report"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "profiler_report_s3_uri = \"s3://{}/{}/rule-output/ProfilerReport/profiler-output\".format(\n",
    "    bucket, restored_estimator.base_job_name\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!aws s3 ls $profiler_report_s3_uri/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!aws s3 cp --recursive $profiler_report_s3_uri ./profiler_report/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.display import display, HTML\n",
    "\n",
    "display(HTML('<b>Review <a target=\"blank\" href=\"./profiler_report/profiler-report.html\">Profiler Report</a></b>'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Review the Profiling Report in SM Studio\n",
    "![SageMaker Studio Extensions](img/studio_pipeline_training_debugger_assigned.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Review the Pipeline in SM Studio\n",
    "![SageMaker Studio Extensions](img/sm_studio_extensions.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Release Resources"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%html\n",
    "\n",
    "<p><b>Shutting down your kernel for this notebook to release resources.</b></p>\n",
    "<button class=\"sm-command-button\" data-commandlinker-command=\"kernelmenu:shutdown\" style=\"display:none;\">Shutdown Kernel</button>\n",
    "\n",
    "<script>\n",
    "try {\n",
    "    els = document.getElementsByClassName(\"sm-command-button\");\n",
    "    els[0].click();\n",
    "}\n",
    "catch(err) {\n",
    "    // NoOp\n",
    "}    \n",
    "</script>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%javascript\n",
    "\n",
    "try {\n",
    "    Jupyter.notebook.save_checkpoint();\n",
    "    Jupyter.notebook.session.delete();\n",
    "}\n",
    "catch(err) {\n",
    "    // NoOp\n",
    "}"
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:081325390199:image/datascience-1.0"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
